package com.demo.common.redis.stream;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;

import com.demo.common.redis.MsgUtil;
import com.demo.common.redis.constants.RedisStream;
import com.demo.common.redis.constants.RedisStreamKey;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;

/**
 * <p>
 * 自定义分组消费者监听
 * </p>
 *
 * @author molong
 * @date 2022/6/22
 */
@Slf4j
abstract class BaseStreamListener<T> implements StreamListener<String, MapRecord<String, String, String>> {

    /**
     * 拿到消息内容的key
     */
    protected static final String PAYLOAD_KEY = "payload";

    /**
     * 消息处理次数key
     */
    protected static final String HANDLE_COUNT_KEY = "handleCount";

    /**
     * 最大处理次数
     */
    private static final long MAX_HANDLE_COUNT = 5;

    /**
     * 获取要订阅的消息key
     * @return 业务上的streamKey， 这不是一个真正的key
     */
    protected abstract String getStreamKey();

    /**
     * redis 操作对象
     */
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 获取 redis 操作对象
     * @return  RedisTemplate<String, String>
     */
    protected RedisTemplate<String, String> getRedisTemplate(){
        if(redisTemplate == null){
            redisTemplate = SpringUtil.getBean(StringRedisTemplate.class);
        }
        return redisTemplate;
    }

    /**
     * 序列化与反序列化对象
     */
    private ObjectMapper objectMapper;

    /**
     * 获取序列化与反序列化器
     * @return ObjectMapper
     */
    protected ObjectMapper getObjectMapper(){
        if(objectMapper == null){
            objectMapper = SpringUtil.getBean(ObjectMapper.class);
        }
        return objectMapper;
    }

    /**
     * 获取反序列化具体的类型
     * <code>
     *     return new TypeReference<TestObject>() {};
     * </code>
     * @return 返回一个反序列化用的类型
     */
    public abstract TypeReference<T> getTypeReference();

    /**
     * 获取完整的streamKey
     * @return 完整的streamKey
     */
    public String getFullStreamKey(){
        return RedisStream.STREAM_KEY_SUFFIX+getStreamKey();
    }

    /**
     * 反序列化
     * @return 反序列化后的对象
     */
    protected T deserialization(MapRecord<String, String, String> message){
        String json = message.getValue().get(PAYLOAD_KEY);
        return MsgUtil.deserialization(json, getObjectMapper(), getTypeReference());
    }

    /**
     * 消息侦听, 如果消息无法正确处理，请抛出异常
     * @param message 消息对象
     */
    public abstract void onMessage(T message);

    /**
     * 默认从消费者中最后消费的消息后开始消费
     */
    private ReadOffset readOffset;

    public void setReadOffset(ReadOffset readOffset){
        this.readOffset = readOffset;
    }

    public ReadOffset getReadOffset() {
        return readOffset;
    }

    /**
     * 提供一个默认的回调函数实现实现, 该实现会在消息消费后删除消息
     * @return 返回函数
     */
    protected java.util.function.Consumer<RecordId> getDefaultDeleteRecordCallback(){
        return recordId -> {
            //消费后的回调函数
            Long delete = getRedisTemplate().opsForStream().delete(getFullStreamKey(), recordId);
            log.info("-------key: {}, 消费完毕，执行消息回调，删除消息id=:{}, 删除的长度为：{}", getFullStreamKey(), recordId, delete);
        };
    }

    /**
     * 返回一个消费完毕后的回调方法
     * @return 回调函数
     */
    protected java.util.function.Consumer<RecordId> getCallback(){
        return null;
    }

    /**
     * 默认的回调函数
     * @param recordId 消息记录id
     */
    protected void callback(RecordId recordId){
        java.util.function.Consumer<RecordId> callback = getCallback();
        if(Objects.nonNull(callback)) {
            callback.accept(recordId);
        }
        //在消费端控制一个key中的最大消息数，如果超过最大消息数，将之前的消息推掉一个，避免消息越积越多
        Long msgSize = getRedisTemplate().opsForStream().size(getFullStreamKey());
        msgSize = Optional.ofNullable(msgSize).orElse(0L);
        if(RedisStream.HISTORY_MAX_LEN < msgSize){
            //超过这个值，则清理掉最后一条消息,即消费一条，同时清理一条已消费记录
            Long trim = getRedisTemplate().opsForStream().trim(getFullStreamKey(), msgSize - 1);
            log.info("-----key:{}, 消费完毕，总消息数：{}, 历史消息阈值：{}， 此时丢弃的旧消息数：{}", getFullStreamKey(), msgSize, RedisStream.HISTORY_MAX_LEN, trim);
        }
    }

    /**
     * 获取一个作者邮箱, 当消费消息发生异常时，会向对应的邮箱发送邮件
     * @return String
     */
    public abstract String getAuthorEmail();

    /**
     * 异常处理
     * @param e 异常对象
     * @param msgId 消息id
     * @param msg 消息内容
     */
    protected void exceptionHandler(Exception e, String msgId, MapRecord<String, String, String> msg){
        log.error("key:{}, 消息处理异常：messageId={}, exception:{}", getFullStreamKey(), msgId, e);
        //消息的消费过程抛出的异常必须要处理，否则会导致容器中的任务状态会变更为取消，此key上的监听将不在收到消息。
        //构建错误日志内容
        StringWriter writer = new StringWriter();
        PrintWriter printWriter = new PrintWriter(writer);
        e.printStackTrace(printWriter);
        printWriter.close();
        ThreadUtil.execute(() -> ErrorProcess.execution(getAuthorEmail(), getFullStreamKey(), writer.toString(), msgId));

        //如果处理次数超过5次都没成功，那么此时将消息丢入死信
        int handleCount = Integer.parseInt(msg.getValue().get(HANDLE_COUNT_KEY));
        if(handleCount > MAX_HANDLE_COUNT){
            //丢入死信队列
            StreamDeadlyMessage streamDeadlyMessage = new StreamDeadlyMessage();
            streamDeadlyMessage.setCreateTime(LocalDateTime.now());
            streamDeadlyMessage.setHandleCount(handleCount);
            streamDeadlyMessage.setKey(getFullStreamKey());
            streamDeadlyMessage.setMsg(deserialization(msg));

            Map<String, String> mapMsg = new HashMap<>(1);
            mapMsg.put(PAYLOAD_KEY, JSONUtil.toJsonStr(streamDeadlyMessage));
            StringRecord stringRecord = StreamRecords.string(mapMsg).withStreamKey(RedisStream.STREAM_KEY_SUFFIX + RedisStreamKey.DEADLY_MSG);
            getRedisTemplate().opsForStream().add(stringRecord);
        }else {
            //处理失败，将消息再次存入队列中, 处理次数加1
            msg.getValue().put(HANDLE_COUNT_KEY, String.valueOf(++handleCount));
            StringRecord stringRecord = StreamRecords.string(msg.getValue()).withStreamKey(getFullStreamKey());
            getRedisTemplate().opsForStream().add(stringRecord);
        }

    }

}
